package org.wenzi.com;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author zhaozuowen
 * @date 2021-10-18 11:42
 */
public class BatchSqlWordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnvironment = BatchTableEnvironment.create(env);

        List<WordCount> data = Stream.of("Es Flink Flink clickHouse".split("\\W+")).map(s -> new WordCount(s,1)).collect(Collectors.toList());
        DataSource<WordCount> dataSet = env.fromCollection(data);
        Table table = tableEnvironment.fromDataSet(dataSet,"word,frequency");
        table.printSchema();

        tableEnvironment.createTemporaryView("WordCount",table);

        Table table02 = tableEnvironment.sqlQuery("select word,sum(frequency) as frequency from WordCount group by word");
        DataSet<WordCount> tuple2DataSet = tableEnvironment.toDataSet(table02, WordCount.class);
        tuple2DataSet.print();


    }


     public  static class WordCount{
        public String word;
         public Integer frequency;

         public WordCount() {
         }

         public WordCount(String word, Integer frequency) {
             this.word = word;
             this.frequency = frequency;
         }

         public String getWord() {
             return word;
         }

         public void setWord(String word) {
             this.word = word;
         }

         public Integer getFrequency() {
             return frequency;
         }

         public void setFrequency(Integer frequency) {
             this.frequency = frequency;
         }

         @Override
         public String toString() {
             return "WordCount{" +
                     "word='" + word + '\'' +
                     ", frequency=" + frequency +
                     '}';
         }
     }


}
